use crate::exchange::lib::{CONFIG, TICKER, Ticker, callback};
use crate::exchange::ourbit::rest::TICKER_STEP;
use crate::utils::{READ_BUFFER_SIZE, WRITE_BUFFER_SIZE, create_text_frame, parse_frame, start_ws};
use sonic_rs::{JsonValueTrait, json, pointer};
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncWriteExt, BufReader, BufWriter};

pub async fn ws(symbols: Arc<Vec<String>>) {
    loop {
        let (rx, tx) = start_ws(
            "futures.ourbit.com",
            Some("/edge"),
            None,
            CONFIG.proxy.as_str(),
        )
        .await;

        let mut buf_reader = BufReader::with_capacity(READ_BUFFER_SIZE, rx);
        let mut buf_writer = BufWriter::with_capacity(WRITE_BUFFER_SIZE, tx);

        for symbol in symbols.iter() {
            let temp_symbol = format!("{}_USDT", &symbol[..symbol.len() - 4]);

            let msg = json!({
                "method": "sub.depth.step",
                "param": {
                    "step": TICKER_STEP.get().unwrap().get(symbol).unwrap(),
                    "symbol": temp_symbol
                }

            });
            let msg = create_text_frame(msg.to_string().as_str());
            buf_writer.write_all(&msg).await.unwrap();
            buf_writer.flush().await.unwrap();
        }

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(10));
            loop {
                interval.tick().await;
                let ping_frame = create_text_frame(json!({"method":"ping"}).to_string().as_str());

                if buf_writer.write_all(&ping_frame).await.is_err() {
                    break;
                }
                if buf_writer.flush().await.is_err() {
                    break;
                }
            }
        });

        loop {
            match parse_frame(&mut buf_reader).await {
                Ok((opcode, payload)) => {
                    match opcode {
                        0x1 => {
                            if (unsafe {
                                sonic_rs::get_from_slice_unchecked(&payload, &["channel"]).unwrap()
                            })
                            .as_str()
                            .unwrap()
                                != "push.depth.step"
                            {
                                continue;
                            }

                            let timestamp = unsafe {
                                sonic_rs::get_from_slice_unchecked(&payload, &["data", "ct"])
                            }
                            .as_i64()
                            .unwrap();

                            let bid = unsafe {
                                sonic_rs::get_from_slice_unchecked(
                                    &payload,
                                    pointer!["data", "bids", 0, 0],
                                )
                            }
                            .as_f64()
                            .unwrap();
                            let bid_amount = unsafe {
                                sonic_rs::get_from_slice_unchecked(
                                    &payload,
                                    pointer!["data", "bids", 0, 1],
                                )
                            }
                            .as_f64()
                            .unwrap();

                            let ask = unsafe {
                                sonic_rs::get_from_slice_unchecked(
                                    &payload,
                                    pointer!["data", "asks", 0, 0],
                                )
                            }
                            .as_f64()
                            .unwrap();
                            let ask_amount = unsafe {
                                sonic_rs::get_from_slice_unchecked(
                                    &payload,
                                    pointer!["data", "asks", 0, 1],
                                )
                            }
                            .as_f64()
                            .unwrap();

                            let symbol = unsafe {
                                sonic_rs::get_from_slice_unchecked(&payload, &["symbol"]).unwrap()
                            }
                            .as_str()
                            .unwrap()
                            .replace("_", "")
                            .to_string();

                            let mut value = TICKER
                                .entry(symbol.clone())
                                .or_insert_with(|| Ticker::default());

                            value.coinex = (bid, ask);
                            value.coinex_timestamp = timestamp;
                            value.coinex_bid_amount = bid_amount;
                            value.coinex_ask_amount = ask_amount;

                            drop(value);

                            callback(symbol).await;
                        }
                        0x2 => {
                            // 二进制帧
                            println!("收到二进制数据，长度: {}", payload.len());
                        }
                        0x8 => {
                            // 关闭帧
                            println!("连接关闭");
                            break;
                        }
                        0x9 => {
                            println!("收到ping帧");
                        }
                        0xa => {
                            println!("收到pong帧");
                        }
                        _ => {
                            println!("收到其他帧，opcode: {opcode}");
                        }
                    }
                }
                Err(e) => {
                    println!("读取错误: {e}");
                    tokio::time::sleep(Duration::from_secs(1)).await;
                    break;
                }
            }
        }
    }
}
